package com.pxene.dmp.mapper;

import java.io.IOException;
import java.net.URLDecoder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import com.pxene.dmp.common.URLUtils;
import com.pxene.dmp.constant.BaseConstant;
import com.pxene.dmp.constant.PatternList;
import com.pxene.dmp.constant.PatternList.Param;
import com.pxene.dmp.constant.PatternList.Pattern;
import com.pxene.dmp.constant.Province;
import com.pxene.dmp.useragent.Parser;

public class ClearDataMapper extends Mapper<LongWritable, Text, LongWritable, Text> {
	
	private Log log = LogFactory.getLog(ClearDataMapper.class);
	
	private MultipleOutputs<LongWritable, Text> mos;
	// useragent解析器
	private Parser parser = new Parser();

	private Text resultVal = new Text();

	private Map<String, String> domains = new HashMap<String, String>();
	private Map<String, String> apis = new HashMap<String, String>();
	private List<String> invalids = new ArrayList<String>();

	public static final class DATA_TYPE {
		public static final int YD_3G = 1;
		public static final int YD_4G = 2;
	};

	public static final String IP_ADDRESS_REGEX = "^[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}$";
	public static final String USER_INFO_REGEX = ".*(\\\\?|&)(idfa=|ida=|androidid=|android_id=|anid=|adrid=|mac=|imei=|deviceid=|aduserid=).*";
	public static final String SEARCH_WORD_REGEX = ".*(\\\\?|&)(q=|word=|query=|kw=|wd=|keyword=).*";
	public static final String WEIXIN_BIZ_REGEX = ".*mp\\.weixin\\.qq\\.com/s\\?__biz=(?=.*?mid=)(?=.*?idx=)(?=.*?sn=).*";
	
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		// 设置配置文件名
		System.setProperty("config-file", conf.get("config-file"));
		for (String domain : conf.getStrings("domains")) {
			String[] array = domain.split("\t");
			if (array.length == 2) {
				String code = array[0];
				String url = array[1];
				domains.put(code, url);
			}
		}
		// 设置灵活接口配置文件名
		for (String api : conf.getStrings("apis")) {
			String[] array = api.split("\\|");
			if (array.length > 2) {
				String code = array[0];
				String role = array[1];
				for (int i=2; i<array.length; i++) {
					String param = "${" + array[i] + "}";
					if (role.contains(param)) {
						role = ".*" + role.replace(param, ".*") + ".*";
					} else {
						role = ".*" + role + "(?=.*?" + array[i] + "=).*";
					}
				}
				apis.put(code, role);
			}
		}
		// 域名过滤配置文件加载
		for (String invalid : conf.getStrings("invalids")) {
			if (invalid.length() > 0) {
				invalids.add(invalid);
			}
		}
		
		mos = new MultipleOutputs<LongWritable, Text>(context);
		log.info(BaseConstant.LOG_PREFIX + "domains amount=" + domains.size());
		log.info(BaseConstant.LOG_PREFIX + "apis amount=" + apis.size());
		log.info(BaseConstant.LOG_PREFIX + "invalids amount=" + invalids.size());
	}


	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		String[] data = value.toString().split(BaseConstant.FILE_SEPARATOR);

		String imsi = null;
		String mdn = null;
		String meid = null;
		String bsid = null;
		String starttime = null;
		String userAgent = null;
		String destinationURL = null;
		String domainName = null;
		int dataType = 0;

		FileSplit fileSplit = (FileSplit) context.getInputSplit();
		String filePath = fileSplit.getPath().toString();

		if (filePath.contains("_4G")) {
			if (data.length != 40) {
				return;
			}
			imsi = data[0].trim();
			mdn = data[1].trim();
			meid = data[2].trim();
			bsid = "JKLMNOPQRSTU";
			// 精确到小时
			starttime = data[17].length() > 10 ? data[17].trim().substring(0, 10) : "";
			userAgent = data[28].trim();
			destinationURL = URLDecoder.decode(data[29].trim().replaceAll("%(?![0-9a-fA-F]{2})", "%25"), "UTF-8");
			domainName = data[30].trim();
			dataType = DATA_TYPE.YD_4G;
		} else {
			if (data.length != 38) {
				return;
			}
			imsi = data[0].trim();
			mdn = data[1].trim();
			meid = data[2].trim();
			bsid = data[12].trim();
			// 精确到小时
			starttime = data[17].length() > 10 ? data[17].trim().substring(0, 10) : "";
			userAgent = data[26].trim();
			destinationURL = URLDecoder.decode(data[27].trim().replaceAll("%(?![0-9a-fA-F]{2})", "%25"), "UTF-8");
			domainName = data[28].trim();
			dataType =DATA_TYPE.YD_3G;
		}
		
		if (StringUtils.isEmpty(imsi) || StringUtils.isEmpty(destinationURL)) {
			return;
		}
		// 根据文件来源，添加省份字段
		String provAbbr = "";
		for (Province province : Province.values()) {
			if (filePath.contains(province.getName())) {
				provAbbr = province.getAbbr();
			}
		}

		// 过滤掉不是以http或https:开头的url
		if (!destinationURL.startsWith("http://") && !destinationURL.startsWith("https://")) {
			return;
		}
		// 过滤掉IP为主机名的url
		if (domainName.matches(IP_ADDRESS_REGEX)) {
			return;
		}

		// 采集userinfo
		if (destinationURL.matches(USER_INFO_REGEX)) {
			resultVal.set(imsi + "|" + mdn + "|" + meid + "|" + domainName + "|" + destinationURL 
					+ "|" + parser.parseDevice(userAgent).family.trim() + "|" + parser.parseOS(userAgent).family.trim() 
					+ "|" + provAbbr + "|" + dataType);
			mos.write(key, resultVal, BaseConstant.HDFS_TEMP_USERINFO_DIR);
		}
		// 采集搜索内容
		if (destinationURL.matches(SEARCH_WORD_REGEX)) {
			resultVal.set(imsi + "|" + bsid + "|" + starttime + "|" + destinationURL + "|" + provAbbr + "|" + dataType);
			mos.write(key, resultVal, BaseConstant.HDFS_TEMP_SEARCHWORD_DIR);
		}
		// 采集微信biz
		if (destinationURL.matches(WEIXIN_BIZ_REGEX)) {
			resultVal.set(imsi + "|" + bsid + "|" + starttime + "|" + destinationURL + "|" + provAbbr + "|" + dataType);
			mos.write(key, resultVal, BaseConstant.HDFS_TEMP_WEIXINBIZ_DIR);
		}
		// 采集电商code
		PatternList businessCodeList = PatternList.load("businesscode.json");
		for (Pattern businessCode : businessCodeList.elements()) {
			if (destinationURL.matches(businessCode.getRegex())) {
				resultVal.set(imsi + "|" + provAbbr + "|" + bsid + "|" + starttime + "|" + businessCode.getCode() + "|" + destinationURL + "|" + dataType);
				mos.write(key, resultVal, BaseConstant.HDFS_TEMP_BUSINESSCODE_DIR);
			}
		}
		// 社交账号
		PatternList contactNumList = PatternList.load("contactnum.json");
		for (Pattern contactNum : contactNumList.elements()) {
			if (destinationURL.matches(contactNum.getRegex())) {
				resultVal.set(imsi + "|" + provAbbr + "|" + bsid + "|" + starttime + "|" + contactNum.getCode() + "|" + destinationURL + "|" + dataType);
				mos.write(key, resultVal, BaseConstant.HDFS_TEMP_BUSINESSCODE_DIR);
			}
		}
		// 坐标
		PatternList locationList = PatternList.load("location.json");
		for (Pattern location : locationList.elements()) {
			if (destinationURL.matches(location.getRegex())) {
				resultVal.set(imsi + "|" + provAbbr + "|" + bsid + "|" + starttime + "|" + location.getCode() + "|" + destinationURL + "|" + dataType);
				mos.write(key, resultVal, BaseConstant.HDFS_TEMP_BUSINESSCODE_DIR);
			}
		}
		// 访问过天气的
		PatternList weatherCityList = PatternList.load("weathercity.json");
		for (Pattern weatherCity : weatherCityList.elements()) {
			if (destinationURL.matches(weatherCity.getRegex())) {
				resultVal.set(imsi + "|" + provAbbr + "|" + bsid + "|" + starttime + "|" + weatherCity.getCode() + "|" + destinationURL + "|" + dataType);
				mos.write(key, resultVal, BaseConstant.HDFS_TEMP_WEATHERCITY_DIR);
			}
		}
		// 查看过汽车信息的
		PatternList autoInfoList = PatternList.load("autoinfo.json");
		for (Pattern autoInfo : autoInfoList.elements()) {
			if (destinationURL.matches(autoInfo.getRegex())) {
				resultVal.set(imsi + "|" + provAbbr + "|" + bsid + "|" + starttime + "|" + autoInfo.getCode() + "|" + destinationURL + "|" + dataType);
				mos.write(key, resultVal, BaseConstant.HDFS_TEMP_AUTOINFO_DIR);
			}
		}
		// 访问过指定域名的imsi
		for (Entry<String, String> entry : domains.entrySet()) {
			if (domainName.startsWith(entry.getValue()) || domainName.contains("." + entry.getValue())) {
				resultVal.set(entry.getKey() + "|" + provAbbr + "|" + imsi + "|" + dataType);
				mos.write(key, resultVal, BaseConstant.HDFS_TEMP_DOMAINUSER_DIR);
			}
		}
		//通过灵活接口获取domainInfo信息
		for (Entry<String, String> entry : apis.entrySet()) {
			log.info(BaseConstant.LOG_PREFIX + "########" + entry.getValue() + "#########");
			if (destinationURL.matches(entry.getValue())) {
				resultVal.set(imsi + "|" + provAbbr  + "|" + bsid + "|" + starttime + "|" + entry.getKey() + "|" + destinationURL + "|" + dataType);
				mos.write(key, resultVal, BaseConstant.HDFS_TEMP_DOMAININFO_DIR);
			}
		}
		
		//通过灵活接口获取ticketInfo信息
		for (Entry<String, String> entry : apis.entrySet()) {
			if(domainName.startsWith("0009") || domainName.startsWith("0008"))
				log.info(BaseConstant.LOG_PREFIX + "########" + entry.getValue() + "#########");
				if (destinationURL.matches(entry.getValue())) {
					resultVal.set(imsi + "|" + provAbbr  + "|" + bsid + "|" + starttime + "|" + entry.getKey() + "|" + destinationURL + "|" + dataType);
					mos.write(key, resultVal, BaseConstant.HDFS_TEMP_TICKETINFO_DIR);
				}
		}
		
		//执行topk
		if(!invalids.contains(domainName)){
			resultVal.set(imsi + "|" + provAbbr + "|" + domainName + "|" + dataType);
			mos.write(key, resultVal, BaseConstant.HDFS_TEMP_FAVOURITESITE_DIR);
		}
	}

	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		mos.close();
	}
	
	
//	public static void main(String[] args) {
//		String url="http://api.app.yiche.com/webapi/api.ashx?method=bit.seriessummary&serialid=3721&cityid=1201&sign=f6f54821dfb46fd942d09320ea8aec08";
//		PatternList autoInfoList = PatternList.load("autoinfo.json");
//		for (Pattern autoInfo : autoInfoList.elements()) {
//			System.out.println(autoInfo.getRegex());
//			if (url.matches(autoInfo.getRegex())) {
//				System.out.println("matched!");
//				Param[] params = autoInfo.getParams();
//				for (Param param : params) {
//					System.out.println(URLUtils.regexpExtract(url, param.getExpr()));
//				}
//			}
//		}
//	}
	
}
